﻿using Akka.Actor;
using Polly;
using System;
using System.Collections.Generic;
using System.Linq;
using System.Threading.Tasks;

namespace Dapper.Sharding.CAP
{
    internal class ServerActor : ReceiveActor
    {
        private ServerCap ServerCap { get; }
        public ServerActor(ServerCap _server)
        {
            ServerCap = _server;

            Receive<TranMessage>(msg =>
            {
                var ok = ServerCap.dict.TryGetValue(msg.Token, out var response);
                if (ok)
                {
                    lock (response._locker)
                    {
                        if (response.Complete) //已经完成了不必再执行
                        {
                            return;
                        }
                        var list = response.TranMessageList;
                        if (!list.Any(a => a.TranId == msg.TranId))
                        {
                            msg.ClientActor = Sender.Path.ToString();
                            list.Add(msg);
                        }
                        //全部都执行完毕了
                        if (list.Count == response.TotalCount)
                        {
                            IEnumerable<TranMessage> msgList;
                            var isCommit = false;
                            if (!list.Any(a => a.TranPreCommit == false))//所有事务都预提交成功
                            {
                                //让所有预提交事务进行真正提交
                                msgList = list;
                                isCommit = true;
                                response.Success = true;
                            }
                            else //存在有提交失败的事务
                            {
                                //获取那些预提交的事务，准备进行回滚
                                msgList = list.Where(w => w.TranPreCommit == true);
                                response.Success = false;
                            }
                            foreach (var item in msgList)
                            {
                                if (response.RetryCount == 0)
                                {
                                    ServerCap.Tell(item.ClientActor, new CommitTellMessage { TranId = item.TranId, Commit = isCommit });
                                }
                                else
                                {
                                    Policy.Handle<Exception>().RetryAsync(response.RetryCount).ExecuteAndCaptureAsync(async () =>
                                    {
                                        var cmsg = new CommitMessage { TranId = item.TranId, Commit = isCommit };
                                        await ServerCap.AskAsync(item.ClientActor, cmsg);
                                    });
                                }

                            }
                            response.Complete = true; //标记完成
                        }
                        //如果一直等不到所有的都执行完毕，那么就一直等待，让客户端自己超时，自己回滚
                    }
                }
            });

        }
    }
}
